package org.apache.flink.runtime.resourceprovider.slot;

import org.apache.flink.api.common.ConsumerID;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.clusterframework.types.SlotID;

import java.util.Objects;

/** Model to save local slot allocation info. */
public class SlotAllocationSnapshot implements java.io.Serializable {

    private static final long serialVersionUID = 1L;

    private final SlotID slotID;
    private final ConsumerID consumerId;
    private final String consumerTargetAddress;
    private final AllocationID allocationId;
    private final ResourceProfile resourceProfile;

    public SlotAllocationSnapshot(
            SlotID slotID,
            ConsumerID consumerId,
            String consumerTargetAddress,
            AllocationID allocationId,
            ResourceProfile resourceProfile) {
        this.slotID = slotID;
        this.consumerId = consumerId;
        this.consumerTargetAddress = consumerTargetAddress;
        this.allocationId = allocationId;
        this.resourceProfile = resourceProfile;
    }

    public SlotID getSlotID() {
        return slotID;
    }

    public ConsumerID getConsumerId() {
        return consumerId;
    }

    public String getConsumerTargetAddress() {
        return consumerTargetAddress;
    }

    public AllocationID getAllocationId() {
        return allocationId;
    }

    public ResourceProfile getResourceProfile() {
        return resourceProfile;
    }

    public int getSlotIndex() {
        return slotID.getSlotNumber();
    }

    @Override
    public String toString() {
        return "SlotAllocationSnapshot{"
                + "slotID="
                + slotID
                + ", consumerId="
                + consumerId
                + ", consumerTargetAddress='"
                + consumerTargetAddress
                + '\''
                + ", allocationId="
                + allocationId
                + ", resourceProfile="
                + resourceProfile
                + '}';
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        SlotAllocationSnapshot that = (SlotAllocationSnapshot) o;
        return slotID.equals(that.slotID)
                && consumerId.equals(that.consumerId)
                && consumerTargetAddress.equals(that.consumerTargetAddress)
                && allocationId.equals(that.allocationId)
                && resourceProfile.equals(that.resourceProfile);
    }

    @Override
    public int hashCode() {
        return Objects.hash(slotID, consumerId, consumerTargetAddress, allocationId, resourceProfile);
    }
}
